package com.o2o.cleaning.month.platform.ebusiness_plat.brand_modular.utils

import com.alibaba.fastjson.JSON
import org.apache.spark.sql.{DataFrame, Row, SparkSession}


class brand_modify_util {

  def brand_modify_1(data: DataFrame, spark: SparkSession, platformName: String, year: Int, month: Int, timeStamp: String): DataFrame = {

    val new_brand = data.withColumnRenamed("brandId", "brandValueId")
      .dropDuplicates("brandValueId")
      .selectExpr("*", "concat(brandValueId,'|',firstCategoryId) as brandCcId", s"'${timeStamp}' as timeStamp",s"'${platformName}' as platform")
//      .withColumnRenamed("platformName", "platform")


    val obs = s"s3a://o2o-dimension-table/brandName_table/${year}/${month}/"
    //    天猫京东苏宁品牌表
    val brandTableAll = spark.read.json(obs + "tmall", obs + "jd", obs + "suning").registerTempTable("brandTableAll")
    //    从品牌结果表中切出中文和英文
    spark.sql(
      """
        |select *,
        |case when brandName_cn rlike "\\/" then concat(firstCategoryId,'_',split(brandName_cn,"\\/")[0]) else concat(firstCategoryId,'_',brandName_cn) end source_cn,
        |case when brandName_cn rlike "\\/" then concat(firstCategoryId,'_',split(brandName_cn,"\\/")[1]) else concat(firstCategoryId,'_',brandName_cn) end source_en
        |from brandTableAll
      """.stripMargin)
      .registerTempTable("sourceTable")

    /**
     * 新增品牌表通过正则进行清洗
     */
    val brandAddData = brandETL(new_brand, spark)
      .drop("brandName_en", "brand_type", "brand_state", "brand_isLaoZiHao")
      .registerTempTable("brandNewTable")

    /**
     * 将清洗完毕的品牌进行切割
     */

    spark.sql(
      """
        |select *,
        |case when brandName_cn rlike "\\/" then concat(firstCategoryId,'_',split(brandName_cn,"\\/")[0]) else concat(firstCategoryId,'_',brandName_cn) end new_cn,
        |case when brandName_cn rlike "\\/" then concat(firstCategoryId,'_',split(brandName_cn,"\\/")[1]) else concat(firstCategoryId,'_',brandName_cn) end new_en
        |from brandNewTable
      """.stripMargin)
      .registerTempTable("newTable")

    //    1、中文和中文进行关联
    val cncnJoin = spark.sql(
      """
        |select
        |a.brandCcId,
        |a.brandName,
        |a.brandValueId,
        |a.firstCategoryId,
        |a.platform,
        |a.timeStamp,
        |b.brandName_cn,
        |b.brandName_en,
        |b.brand_state,
        |b.brand_type,
        |b.brand_isLaoZiHao
        |from newTable a
        |join
        |sourceTable b
        |on a.new_cn = b.source_cn
      """.stripMargin)

    //    2、中文和中文关联不上的
    val cncnNotJoin = spark.sql(
      """
        |select a.*,
        |b.source_cn,
        |b.source_en,
        |b.brandName_cn,
        |b.brandName_en,
        |b.brand_state,
        |b.brand_type,
        |b.brand_isLaoZiHao
        |from newTable a
        |left join
        |sourceTable b
        |on a.new_cn = b.source_cn
        |where b.source_cn is null
      """.stripMargin)
    cncnNotJoin.registerTempTable("cncnNotJoin")
    //    3、中文和英文关联
    val cnenJoin = spark.sql(
      """
        |select
        |a.brandCcId,
        |a.brandName,
        |a.brandValueId,
        |a.firstCategoryId,
        |a.platform,
        |a.timeStamp,
        |b.brandName_cn,
        |b.brandName_en,
        |b.brand_state,
        |b.brand_type,
        |b.brand_isLaoZiHao
        |from cncnNotJoin a
        |join
        |sourceTable b
        |on a.new_cn = b.source_en
      """.stripMargin)


    //    4、中文和英文关联不上的
    val cnenNotJoin = spark.sql(
      """
        |select
        |a.*,
        |b.source_cn,
        |b.source_en,
        |b.brandName_cn,
        |b.brandName_en,
        |b.brand_state,
        |b.brand_type,
        |b.brand_isLaoZiHao
        |from cncnNotJoin a
        |left join
        |sourceTable b
        |on a.new_cn = b.source_en
        |where b.source_en is null
      """.stripMargin).registerTempTable("cnenNotJoin")
    //    5、英文和英文关联
    val enenJoin = spark.sql(
      """
        |select
        |a.brandCcId,
        |a.brandName,
        |a.brandValueId,
        |a.firstCategoryId,
        |a.platform,
        |a.timeStamp,
        |b.brandName_cn,
        |b.brandName_en,
        |b.brand_state,
        |b.brand_type,
        |b.brand_isLaoZiHao
        |from cnenNotJoin a
        |join
        |sourceTable b
        |on a.new_en = b.source_en
      """.stripMargin)


    //    6、英文关联不上的返回原来名称
    val enenNotJoin = spark.sql(
      """
        |select a.*
        |from cnenNotJoin a
        |left join
        |sourceTable b
        |on a.new_en = b.source_en
        |where b.source_en is null
      """.stripMargin)
    val enenNotJoinResult = brandETL(enenNotJoin, spark)

    /**
     * 将新品牌表和老品牌表进行合并
     * cncnJoin: 中文和中文进行关联
     * cnenJoin: 中文和英文关联
     * enenJoin: 英文和英文关联
     * enenNotJoinResult：英文关联不上的使用正则再次清洗
     */

    if (enenNotJoinResult.count() == 0) {
      val brandResult = cncnJoin
        .select("brandValueId", "firstCategoryId", "brandName", "brandName_cn", "brandCcId", "brand_isLaoZiHao", "brand_type", "brand_state", "platform", "timeStamp")
        .union(cnenJoin
          .select("brandValueId", "firstCategoryId", "brandName", "brandName_cn", "brandCcId", "brand_isLaoZiHao", "brand_type", "brand_state", "platform", "timeStamp"))
        .union(enenJoin
          .select("brandValueId", "firstCategoryId", "brandName", "brandName_cn", "brandCcId", "brand_isLaoZiHao", "brand_type", "brand_state", "platform", "timeStamp"))
        .dropDuplicates("brandCcId")
        .drop("brandName_en")
      return brandResult
    } else {
      val brandResult = cncnJoin
        .select("brandValueId", "firstCategoryId", "brandName", "brandName_cn", "brandCcId", "brand_isLaoZiHao", "brand_type", "brand_state", "platform", "timeStamp")
        .union(cnenJoin
          .select("brandValueId", "firstCategoryId", "brandName", "brandName_cn", "brandCcId", "brand_isLaoZiHao", "brand_type", "brand_state", "platform", "timeStamp"))
        .union(enenJoin
          .select("brandValueId", "firstCategoryId", "brandName", "brandName_cn", "brandCcId", "brand_isLaoZiHao", "brand_type", "brand_state", "platform", "timeStamp"))
        .union(enenNotJoinResult
          .selectExpr("brandValueId", "firstCategoryId", "brandName", "brandName_cn", "brandCcId", "'0' as brand_isLaoZiHao", "'-1' as brand_type", "'0' as brand_state", "platform", "timeStamp"))
        .dropDuplicates("brandCcId")
        .drop("brandName_en")
      return brandResult
    }


  }

  def brandETL(data: DataFrame, spark: SparkSession): DataFrame = {
    val brandETLResult = data.toJSON.rdd.map(line => {
      val jsons = JSON.parseObject(line)
      val brandName = jsons.getOrDefault("brandName", "-1").toString
        .replaceAll("\\“", "")
        .replaceAll("\\”", "")
        .replaceAll("【", "")
        .replaceAll("】", "")
        .replaceAll("\\[", "")
        .replaceAll("\\]", "")
        .replaceAll("&", "&")
        .replaceAll("（", "(")
        .replaceAll("）", ")")
        .replaceAll("＆", "&")
        .replaceAll("、", "")
        .replaceAll("#", "")
        .replaceAll(";", "")
        .replaceAll("；", "")
        .replaceAll("男鞋", "")
        .replaceAll("女鞋", "")
        .replaceAll("童鞋", "")
        .replaceAll("男包", "")
        .replaceAll("女包", "")
        .replaceAll("单肩包", "")
        .replaceAll("食品", "")
        .replaceAll("罐头", "")
        .replaceAll("酸奶", "")
        .replaceAll("牛奶", "")
        .replaceAll("餐饮具", "")
        .replaceAll("办公用品", "")
        .replaceAll("居家日用", "")
        .replaceAll("鲜花", "")
        .replaceAll("车灯", "")
        .toUpperCase()

      //    只有中文
      if (brandName.isEmpty || brandName.replaceAll("/", "").length == 0) {
        jsons.put("brandName_cn", "其他/OTHER")
      } else if (brandName.replaceAll("""[0-9\u4e00-\u9fa5 ]+""", "").length == 0) {
        //        println("只有中文：" + brandName)
        jsons.put("brandName_cn", brandName)
        //    只有英文
      } else if (brandName.replaceAll("""[0-9A-Z&-' ]+""", "").length == 0) {
        //        println("只有英文：" + brandName)
        jsons.put("brandName_cn", brandName)
        //   只有中文和英文
      } else if (brandName.replaceAll("""[0-9\u4e00-\u9fa5 ]+""", "")
        .replaceAll("""[0-9A-Z&-' ]+""", "").length == 0) {
        val cn = brandName.replaceAll("""[0-9A-Z&-']+""", "").replaceAll("/", "").trim
        val en = brandName.replaceAll("""[\u4e00-\u9fa5]+""", "").replaceAll("/", "").trim
        if (cn.length == 0) {
          //          println("只有中文和英文：英文："+en)
          jsons.put("brandName_cn", en)
        } else if (en.length == 0) {
          //          println("只有中文和英文：中文："+cn)
          jsons.put("brandName_cn", cn)
        } else {
          //          println("只有中文和英文：中文+英文："+cn +"/"+en)
          jsons.put("brandName_cn", cn + "/" + en)
        }
        //    只有中文和英文并且带斜杠
      } else if (brandName.replaceAll("""[0-9\u4e00-\u9fa5 ]+""", "")
        .replaceAll("""[0-9A-Z&-' ]+""", "")
        .replaceAll("/", "")
        .length == 0) {
        val cn = brandName.replaceAll("""[0-9A-Z&-']+""", "").replaceAll("/", "").trim
        val en = brandName.replaceAll("""[\u4e00-\u9fa5]+""", "").replaceAll("/", "").trim
        if (cn.length == 0) {
          //          println("只有中文和英文并且带斜杠：英文："+en)
          jsons.put("brandName_cn", en)
        } else if (en.length == 0) {
          //          println("只有中文和英文并且带斜杠：中文："+cn)
          jsons.put("brandName_cn", cn)
        } else {
          //          println("只有中文和英文并且带斜杠：中文+英文："+cn +"/"+en)
          jsons.put("brandName_cn", cn + "/" + en)
        }
        //  只有中文和英文并且带括号
      } else if (brandName.replaceAll("""[0-9\u4e00-\u9fa5 ]+""", "")
        .replaceAll("""[0-9A-Z&-' ]+""", "")
        .replaceAll("\\(", "")
        .replaceAll("\\)", "")
        .length == 0) {
        val cn = brandName.replaceAll("""[0-9A-Z&-']+""", "")
          .replaceAll("/", "")
          .replaceAll("\\(", "")
          .replaceAll("\\)", "")
          .trim
        val en = brandName.replaceAll("""[\u4e00-\u9fa5]+""", "")
          .replaceAll("/", "")
          .replaceAll("\\(", "")
          .replaceAll("\\)", "")
          .trim
        //      判断数据中是否有英文
        if (en.length == 0) {
          //          println("只有中文和英文并且带括号：中文+英文："+cn)
          jsons.put("brandName_cn", cn)
        } else if (cn.length == 0) {
          //          println("只有中文和英文并且带括号：中文+英文："+en)
          jsons.put("brandName_cn", en)
        } else {
          //          println("只有中文和英文并且带括号：中文+英文："+cn+"/"+en)
          jsons.put("brandName_cn", cn + "/" + en)
        }
        //      这个是中文+英文+斜杠+括号
      } else if (brandName.replaceAll("""[0-9\u4e00-\u9fa5 ]+""", "")
        .replaceAll("""[0-9A-Z&-' ]+""", "")
        .replaceAll("/", "")
        .replaceAll("\\(", "")
        .replaceAll("\\)", "").length == 0) {

        val cn = brandName.replaceAll("""[0-9A-Z&-']+""", "")
          .replaceAll("/", "")
          .replaceAll("\\(", "")
          .replaceAll("\\)", "")
          .trim
        val en = brandName.replaceAll("""[\u4e00-\u9fa5]+""", "")
          .replaceAll("/", "")
          .replaceAll("\\(", "")
          .replaceAll("\\)", "")
        if (cn.length == 0) {
          //          println("中文+英文+斜杠+括号：英文:",en)
          jsons.put("brandName_cn", en)
        } else if (en.length == 0) {
          //          println("中文+英文+斜杠+括号：中文：",en)
          jsons.put("brandName_cn", cn)
        } else {
          //          println("中文+英文+斜杠+括号：中文+英文：",cn+"/"+en)
          jsons.put("brandName_cn", cn + "/" + en)
        }
      } else {
        val cn = brandName.replaceAll("""[0-9A-Z&-'][-.]+""", "")
          .replaceAll("/", "")
          .replaceAll("\\(", "")
          .replaceAll("\\)", "")
          .trim()
        val en = brandName.replaceAll("""[\u4e00-\u9fa5]+""", "")
          .replaceAll("/", "")
          .replaceAll("\\(", "")
          .replaceAll("\\)", "")
          .trim
        if (cn.length == 0) {
          jsons.put("brandName_cn", en)
        } else if (en.length == 0) {
          jsons.put("brandName_cn", cn)
        } else if (cn.equals("-1") || cn.equals("·") || cn.equals("-") || cn.equals("+") || cn.equals(".") || cn.equals("の")) {
          //          println("只有中文和英文：中文是-1："+"其他/OTHER")
          jsons.put("brandName_cn", "其他/OTHER")
        } else if (en.equals("-1") || en.equals("·") || en.equals("-") || en.equals("+") || en.equals(".") || en.equals("の")) {
          //          println("只有中文和英文：英文是-1："+"其他/OTHER")
          jsons.put("brandName_cn", "其他/OTHER")
        } else if (cn.contains("+") || en.contains("+") || cn.contains("-") ||
          en.contains("-") || cn.contains(".") || en.contains(".") || cn.contains("·") || en.contains("·") ||
          cn.contains("の") || en.contains("の")) {
          //          println("中文带+：",cn)
          jsons.put("brandName_cn", brandName)
        } else {
          //                    println("只有中文和英文：中文+英文："+cn +"/"+en)
          jsons.put("brandName_cn", cn + "/" + en)
        }
      }


      jsons.toString
    })
    spark.read.json(brandETLResult)
  }

}
